package com.citusdata.elven.linearRoad;

import com.citusdata.elven.datastore.protobuf.Datastore.*;
import com.citusdata.elven.datastore.DatastoreCoordinator;
import com.citusdata.elven.httpRpc.HttpRpcChannel;
import com.citusdata.elven.httpRpc.HttpRpcChannelManager;
import com.citusdata.elven.httpRpc.HttpRpcController;
import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import com.citusdata.elven.linearRoad.LinearRoadService;
import com.citusdata.elven.router.Endpoint;
import com.citusdata.elven.router.ReplicationGroupRouter;
import com.google.protobuf.RpcCallback;
import java.util.List;


// Coordinator class that handles queries in a distributed setting. For each
// query, this class contacts the related endpoints (hostname and port), and
// receives answers from them. The class then de-duplicates and merges these
// answers. An example query-cycle is as follows:
//
// (1) reportPosition() is called. Based on the expressway and the direction in
// the query, the associated partition (replication group) is calculated. All
// endpoints in the replication group are queried, answers are received, and the
// most accurate answer is selected.
// (2) If the most accurate answer contains a toll assessment, this assessment
// is persisted in the distributed datastore. This time, vehicleId is used as
// the partition key. In a typical Linear Road benchmark run, 20% of all
// position reports generate a toll assessment.
// (3) computeAccount() is called at a later time. Since toll assessments were
// persisted using vehicleId as the partition key, the same key is used when
// retrieving assessments. DatastoreCoordinator de-duplicates and merges these
// assessment, and final computations are performed in this class.


public class LinearRoadCoordinator {
  private final ReplicationGroupRouter router;
  private final HttpRpcChannelManager  channelManager;
  private final DatastoreCoordinator   remoteDatastores; 

  public LinearRoadCoordinator(ReplicationGroupRouter router,
                               HttpRpcChannelManager  channelManager) {
    this.router = router;
    this.channelManager = channelManager;
    this.remoteDatastores = new DatastoreCoordinator(router, channelManager);
  }


  public int hashCode(int expressway, int direction) {
    int result = 17;
    result = 37*result + expressway;
    result = 37*result + direction;
    return result;
  }


  public PositionResponse reportPosition(PositionRequest request) {
    long timestamp = (long) request.getTime();
    List activationTimes = (List) router.getActivationTimes(timestamp, timestamp);
    assert (activationTimes.size() == 1);
    
    int  hashCode = hashCode(request.getXway(), request.getDir());
    long activationTime = (Long) activationTimes.get(0);
    List endpoints = (List) router.getEndpoints(hashCode, activationTime);

    // Make calls to all endpoints, and choose the response with the highest
    // event count.
    int highestEventCount = 0;
    PositionResponse highestResponse = null;

    for (int ep = 0; ep < endpoints.size(); ++ep) {
      Endpoint endpoint = (Endpoint) endpoints.get(ep);
      String hostname = endpoint.getHostname();
      int port = endpoint.getPort();

      HttpRpcChannel httpChannel = channelManager.getChannel(hostname, port);
      HttpRpcController httpController = new HttpRpcController();

      LinearRoadServiceInterface service = LinearRoadServiceInterface.newStub(httpChannel);
      LinearRoadCallback done = new LinearRoadCallback();

      service.reportPosition(httpController, request, done);

      channelManager.releaseChannel(httpChannel);
      
      if (httpController.failed()) {
        router.recordFailure(endpoint);
      } else {
        PositionResponse response = done.positionResponse();
        if (response.getEventCount() >= highestEventCount) {
          highestEventCount = response.getEventCount();
          highestResponse   = response;
        }
      }
    }  // for endpoint index

    // We now have the most accurate response. If it contains a toll assessment
    // from the previous segment, store the toll in the distributed data store.
    if (highestResponse != null && highestResponse.hasAssessment()) {

      TollAssessment toll = highestResponse.getAssessment();
      String vehicleId = Integer.toString(toll.getVid());

      PutRequest.Builder putBuilder = PutRequest.newBuilder();
      putBuilder.setPartitionKey(vehicleId);
      putBuilder.setTimestamp(timestamp);
      putBuilder.setData(toll.toByteString());
      PutRequest putRequest = putBuilder.build();

      remoteDatastores.put(putRequest);
    }

    return highestResponse;
  }


  public AccountResponse computeAccount(AccountRequest accReq) {
    int epochStart = 0;
    List tolls = getHistoricalTolls(accReq.getVid(), epochStart, accReq.getTime());

    AccountResponse response = LinearRoadService.computeAccount(accReq, tolls);
    return response;
  }


  public ExpenditureResponse computeExpenditure(ExpenditureRequest expReq) {
    // Time for historical queries are represented as "negative days"
    int adjustedDay = expReq.getDay() * -1;
    List tolls = getHistoricalTolls(expReq.getVid(), adjustedDay, adjustedDay);

    ExpenditureResponse response = LinearRoadService.computeExpenditure(expReq, tolls);
    return response;
  }

  
  private List getHistoricalTolls(int vehicleId, int startTime, int endTime) {
    String vehicleKey = Integer.toString(vehicleId);

    GetRangeRequest.Builder getBuilder = GetRangeRequest.newBuilder();
    getBuilder.setPartitionKey(vehicleKey);
    getBuilder.setStartTime(startTime);
    getBuilder.setEndTime(endTime);
    GetRangeRequest getRequest = getBuilder.build();

    GetRangeResponse getResponse = remoteDatastores.getRange(getRequest);
    List putRequests = getResponse.getPutRequestsList();

    List historicalTolls = LinearRoadService.parseHistoricalTolls(putRequests);
    return historicalTolls;
  }


  private class LinearRoadCallback implements RpcCallback {
    PositionResponse response;

    public void run(Object object) {
      response = (PositionResponse) object;
    }

    public PositionResponse positionResponse() {
      return response;
    }
  }
}
